package com.lmq

//import com.lmq.filterLen.spark
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.types.{DoubleType, IntegerType, StringType, StructField, StructType}

import java.util
import scala.collection.mutable

/**
 *  TODO: not used
 */
object genSession {

  def genRandomUserId():util.ArrayList[Int] = {
    var x: util.ArrayList[Int] = new util.ArrayList[Int]()
    x.add(1)
    x
  }
  Logger.getLogger("org.apache.spark")
    .setLevel(Level.WARN)

  val spark = SparkSession.builder()
    .master("local[*]")
    .appName("Test")
    .getOrCreate()


  def main(args: Array[String]): Unit = {
    val utils = new utils()
    //CSV schema
    val schema = StructType(
      Array(
        StructField("SessionId", IntegerType, nullable = true),
//        StructField("TimeStr", StringType, nullable = true),
        StructField("ItemId", IntegerType, nullable = true),
        StructField("Context", StringType, nullable = true),
        StructField("Time",DoubleType, nullable = false)
//        StructField("TimeRank",IntegerType, nullable = true)
      )
    )
    //    read CSV
    val v: DataFrame = spark.read
      .schema(schema)
//      .option("header",value = false)
//      .option("header",value = true)
//      .csv("file:///home/iptv/yoochoose/Reslt.csv/1over64.csv")
        .csv("file:///home/iptv/yoochoose/OneOsixfour_v2/part-00000-a24d0f25-8b8d-4e48-bacc-e60f0343ed96-c000.csv")
//        .csv("D:\\pythonProject\\pythonProject\\ComparisonWithYoochoose\\src\\head1.csv")
    //      df.show(false)

    //      val timeTransfer = udf((x:String)=>  utils.timeStr2Tsp(x) )
    //      df.select(
    //        col("SessionId"),
    //        col("TimeStr"),
    //        col("ItemId"),
    //        col("Context"),
    //        timeTransfer( col("TimeStr") ).alias("Time")
    //      )
    //        .createOrReplaceTempView("allData")
    //
    //      val v: DataFrame = spark.sql(""" select * ,rank() over (partition by SessionId order by Time) as
    //    `TimeRank` from allData""")
    v.printSchema()



    println("=====================================")
    v.show(false)
    def sortwithPositionElement(Pt:Int) = udf((x:mutable.WrappedArray[Double])=>  x(Pt) )
    //      println(v.count())
    v.createTempView("v")

    val w = v.groupBy(col("SessionId")).agg(collect_list("ItemId").as("ItemId"), collect_list("Time").as("Time"))
//    w.printSchema()
    val toStrFnc = udf((x:mutable.WrappedArray[AnyVal])=>x.mkString(","))
    w.show(100,truncate = false)
    w.orderBy(sortwithPositionElement(0)(col("Time")))
      .select(
        col("SessionId"),
        toStrFnc(col("ItemId"))

      )
      .coalesce(1).write
      .option("header","false")
      .csv("file:///home/iptv/yoochoose/OneOsixfour_sessions")
    println("Done.")
//      .show(100,false)



            //        .count())
    //        .show(1000,truncate = false)
    //      println(
    // Random generated userid
//
//    spark.sql(
//      """select v.* from
//        |(select ItemId from
//        |( select ItemId, count(1) as cnt
//        |from v
//        |group by ItemId) A where cnt>=5) B
//        |left join v
//        |on B.ItemId=v.ItemId
//        |""".stripMargin)
//    //        .count())
//    //        .show(1000,truncate = false)
//
//    //      val xtable =
//    spark.sql(
//      """select w.SessionId,w.ItemId,w.Context,w.Time,
//        |rank() over (partition by SessionId order by Time) as TimeRank
//        |from
//        |(select v.* from
//        |(select ItemId from
//        |( select ItemId, count(1) as cnt
//        |from v
//        |group by ItemId) A where cnt>=5) B
//        |left join v
//        |on B.ItemId=v.ItemId) w
//        |""".stripMargin).createTempView("xtable")
//    //        .show(false)
//    spark.sql(
//      """
//        |select xtable.SessionId,xtable.ItemId,xtable.Context,xtable.Time
//        |from
//        |(select SessionId from
//        |(select SessionId,sum(1) as cnts from xtable group by SessionId) groupedtable where cnts>1
//        |) ltable
//        |left join xtable
//        |on ltable.SessionId=xtable.SessionId
//        |""".stripMargin).coalesce(1).write
//      .option("header","true")
//      .csv("file:///home/iptv/yoochoose/OneOsixfour")
//    println("Done.")
    //        .show(1000,false)


    // remove sessions which the length is less than 1 and apearance is less than 5.

    //    v.coalesce(1).write
    //      .option("header","true")
    //      .csv("file:///home/iptv/yoochoose/Reslt.csv")

    //      .toPandas()\
    //    .to_csv("helloResult.csv",index=None)




  }

}
